package com.wudl.hudi.source;

import com.wudl.hudi.entity.FlinkCdcBean;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author ：wudl
 * @date ：Created in 2022-02-19 15:08
 * @description：
 * @modified By：
 * @version: 1.0
 */

public class MysqlJdbcSink extends RichSinkFunction<FlinkCdcBean> {
    // 声明连接和预编译语句
    Connection connection = null;
    PreparedStatement insertStmtFlink_cdc = null;
    PreparedStatement insertStmtFlink_cdd = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection("jdbc:mysql://192.168.1.162:3306/wudldb?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai", "root", "123456");
        insertStmtFlink_cdc = connection.prepareStatement("INSERT INTO `wudldb`.`Flink_cdc`(NAME,age,birthday,ts) VALUES(?,?,?,?) ");
        insertStmtFlink_cdd = connection.prepareStatement("INSERT INTO  `wudldb`.`Flink_cdd` (phone,address,ts) VALUES(?,?,?)  ");

    }

    // 每来一条数据，调用连接，执行sql
    @Override
    public void invoke(FlinkCdcBean fc, Context context) throws Exception {
/*****************    */
        insertStmtFlink_cdc.setString(1, fc.getName());
        insertStmtFlink_cdc.setInt(2, fc.getAge());
        insertStmtFlink_cdc.setString(3, fc.getTs().toString());
        insertStmtFlink_cdc.setString(4, fc.getTs().toString());
        insertStmtFlink_cdc.execute();

        insertStmtFlink_cdd.setString(1, fc.getPhone());
        insertStmtFlink_cdd.setString(2, fc.getAddress());
        insertStmtFlink_cdd.setString(3, fc.getTs().toString());
        insertStmtFlink_cdd.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        insertStmtFlink_cdc.close();
        insertStmtFlink_cdd.close();
        connection.close();
    }
}
